package test;

import java.io.IOException;
import java.util.Scanner;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.FileOutputFormat;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.Mapper;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reporter;
import org.apache.hadoop.mapred.SequenceFileInputFormat;
import org.apache.hadoop.mapred.lib.MultipleInputs;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import utils.Utils;

public class CheckColumnCount extends Configured implements Tool{
	public static class Map extends MapReduceBase implements Mapper<BytesWritable, Text, Text, Text> {
		private Text k = new Text();
		enum Counters{INVALID_COLUMN_COUNT};
		public void map(BytesWritable key, Text value, 
				OutputCollector<Text, Text> output, Reporter reporter) throws IOException {
			String[] values = value.toString().split(Utils.auctionSeparator);
			if (values.length < 34) {
				reporter.incrCounter(Counters.INVALID_COLUMN_COUNT, 1);
				k.set(String.valueOf(values.length));
				output.collect(k, value);				
			}
			
		}
	}
	public int run(String[] args) throws Exception {
		// input
		String itemFile = "/group/taobao/taobao/hive/r_auction_auctions/pt=" + args[0] + "000000/if_online=1";
		
		// Output
		String output = Utils.rootDir + args[0] + "/auctioncolumns";
		System.out.println("start job CheckColumnCount...");
		System.out.println("itemFile: " + itemFile);
		System.out.println("output: " + output);
		System.out.println("Continue(0/1)?");
		Scanner scanner = new Scanner(System.in);
		int c = scanner.nextInt(); 
		if (c == 0)
			return -1;
		
		JobConf conf = new JobConf(getConf(), CheckColumnCount.class);
		conf.setJobName("CheckColumnCount");
		conf.setOutputKeyClass(Text.class);
		conf.setOutputValueClass(Text.class);

		MultipleInputs.addInputPath(conf, 
				new Path(itemFile), 
				SequenceFileInputFormat.class, 
				Map.class);

		FileOutputFormat.setOutputPath(conf, new Path(output));
		
		conf.setNumReduceTasks(0);
		JobClient.runJob(conf);
		return 0;
	}

	public static void main(String[] args) throws Exception {
		if (args.length != 1) {
			System.out.println("Usage: hadoop jar xx.jar <date>.");
			System.exit(-1);
		}
		int res = ToolRunner.run(new Configuration(), new CheckColumnCount(),
				args);
		System.exit(res);
	}
}
